use std::sync::Arc;

use tokio::net::TcpSocket;

use crate::{
    backend::{peer::Peer, BackendsMgr},
    error::*,
    http::{base::conn::HttpConn, pool::client_pool::ClientConnPools},
    stream::Stream,
};

use super::{session::ClientSession, HttpProxy};

#[allow(dead_code)]
#[derive(Default)]
pub struct HttpUpstreamStatus {
    // pub peer: Option<Arc<Peer>>,
    pub error: Option<BoxError>,
}

#[allow(dead_code)]
#[derive(Default)]
pub struct HttpUpstreamCtl {
    // 如果proxy不可以重试, 该标记将被设置.
    pub no_retries_allowed: bool,
    pub status_vec: Vec<HttpUpstreamStatus>,
}

impl HttpUpstreamCtl {
    // 如果status_vec为空, 将导致panic.
    pub fn get_last_status(&mut self) -> &mut HttpUpstreamStatus {
        // assert!(!self.status_vec.is_empty());
        let len = self.status_vec.len();
        &mut self.status_vec[len - 1]
    }
}

#[allow(dead_code)]
impl HttpProxy {
    pub async fn connect_to_upstream(&mut self) -> Result<ClientSession> {
        let conf = self.conf.as_ref().unwrap();
        let backend_conf = &conf.backend_runtime_conf;

        if let Some(global_backend_name) = &backend_conf.global_backend_name {
            if !global_backend_name.is_empty() {
                let backend_name = global_backend_name.clone();
                return self
                    .connect_peer_by_global_backend(&backend_name)
                    .await;
            }
        }

        self.connect_peer_by_backend().await
    }

    async fn connect_peer_by_global_backend(
        &mut self,
        backend_name: &str,
    ) -> Result<ClientSession> {
        let backend = match BackendsMgr::get_backend(backend_name) {
            Some(b) => b,
            None => {
                return Error::e_explain(
                    ErrorType::NotFoundBackend,
                    "while in connect_peer_by_global_backend",
                );
            }
        };

        let peer = backend.get_peer(None).await?;
        self.client_hc_key = Some(peer.address.to_string());
        self.connect_peer(peer).await
    }

    async fn connect_peer_by_backend(&mut self) -> Result<ClientSession> {
        let backend = match self
            .conf
            .as_ref()
            .unwrap()
            .backend_runtime_conf
            .backend
            .as_ref()
        {
            Some(b) => b,
            None => {
                return Error::e_explain(
                    ErrorType::NotFoundBackend,
                    "while in connect_peer",
                )
            }
        };
        let peer = backend.get_peer(None).await?;
        self.client_hc_key = Some(peer.address.to_string());
        self.connect_peer(peer).await
    }

    async fn connect_peer(&mut self, peer: Arc<Peer>) -> Result<ClientSession> {
        if let Some(hc) =
            ClientConnPools::get_conn(self.client_hc_key.as_ref().unwrap())
        {
            let core_conf =
                self.conf.as_ref().unwrap().core_runtime_conf.clone();
            return Ok(ClientSession::new_http1(hc, Some(core_conf)));
        }

        let tcp_socket = if peer.address.is_ipv4() {
            TcpSocket::new_v4()
        } else {
            TcpSocket::new_v6()
        }
        .if_err(
            ErrorType::CreateError,
            format!("while in create tcp socket, address: {}", peer.address),
        )?;

        let tcp_stream = tcp_socket.connect(peer.address).await.if_err(
            ErrorType::ConnectError,
            format!("connect peer failed, peer address: {}", peer.address),
        )?;
        let stream: Stream = tcp_stream.into();

        let hc = Box::new(HttpConn::new_with_stream(stream, None));
        let core_conf = self.conf.as_ref().unwrap().core_runtime_conf.clone();
        Ok(ClientSession::new_http1(hc, Some(core_conf)))
    }

    pub fn allow_retry(&self) -> bool {
        if self.upstream_ctl.no_retries_allowed {
            return false;
        }

        // TODO:
        false
    }
}
